﻿using System;
using System.Collections.Generic;
using System.Linq;
using MongoDB.Bson;
using MongoDB.Driver;
using Nest;
using Elasticsearch.Net;
using Polly;
using Serilog;
using HYService.RabbitMQ;
using Elasticsearch.Synchronize.Job.Settings;
using Elasticsearch.Synchronize.Job.Entity;

namespace Elasticsearch.Synchronize.Job.Job
{
    public class EventJoinJob : BaseProcessor<EventJoin>
    {
        private readonly EventJoinSettings _settings = null;

        public EventJoinJob(EventJoinSettings settings) : base(settings)
        {
            _settings = settings;
            //this.Test();
        }

        private void Test()
        {
            int i = 0;
            while (i < 10)
            {
                HYService.RabbitMQ.RabbitHub.Setup("tracejoinexchange", ExchangeType.Fanout,
                    _settings.TailExtraFieldQueue);
                HYService.RabbitMQ.RabbitHub.Broadcast<string>("tracejoinexchange", "552352da5e5b9b103c67cb2f");

                HYService.RabbitMQ.RabbitHub.Setup("syncjoinbyeventidexchange", ExchangeType.Fanout,
                    _settings.SyncJoinByEventIdQueue);
                HYService.RabbitMQ.RabbitHub.Broadcast<string>("syncjoinbyeventidexchange", "10971002");

                i++;
            }
        }

        public void TailExtraField()
        {
            RabbitHub.Subscribe<string>(_settings.TailExtraFieldQueue, (_id, consumer, args) =>
            {
                try
                {
                    Log.Logger.Information(string.Format("received join._id [{0}]", _id));
                    var filter = Builders<EventJoin>.Filter.Eq("_id", ObjectId.Parse(_id));
                    var projection =
                        new BsonDocument {{"_id", 1}, {"JoinId", 1}, {"EventId", 1}, {"ExtraFieldsForES", 1}};
                    var join = _mongoCollection.Find(filter).Project<EventJoin>(projection).FirstOrDefault();
                    if (join != null && join.ExtraFieldsForMongo != null && join.ExtraFieldsForMongo.Any())
                    {
                        var result = Policy.HandleResult<IGetResponse<EventJoin>>(r => !r.IsValid || !r.Found)
                            .WaitAndRetry(
                                4,
                                retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (a, b, c, d) =>
                                {
                                    if (c > 2)
                                    {
                                        Log.Logger.Error("{@model}", new
                                        {
                                            Id = _id,
                                            Exception = new Exception(a.Result.DebugInformation)
                                        });
                                    }

                                    var _fullFilter =
                                        Builders<BsonDocument>.Filter.Eq("_id", ObjectId.Parse(_id));
                                    var _projection = Builders<BsonDocument>.Projection.Exclude("ExtraFields")
                                        .Exclude("ExtraFieldsForES").Exclude("_id");
                                    var _entity = _fullMongoCollection.Find(_fullFilter).Project(_projection)
                                        .FirstOrDefault<BsonDocument>();
                                    if (_entity != null)
                                    {
                                        var entity = _entity.ToDictionary();
                                        var _response = _elasticClient.Index(entity,
                                            s => s.Index(_settings.ElasticsearchIndex).Type(_settings.ElasticsearchType)
                                                .Id(_id).Refresh(Refresh.True));

                                        if (c > 2 && !_response.IsValid)
                                        {
                                            Log.Logger.Error("{@model}", new
                                            {
                                                Id = _id,
                                                Exception = new Exception(_response.DebugInformation)
                                            });
                                        }
                                    }
                                }
                            ).Execute(() =>
                                {
                                    var one = _elasticClient.Get<EventJoin>(_id,
                                        s => s.Index(_settings.ElasticsearchIndex).Type(_settings.ElasticsearchType));

                                    return one;
                                }
                            );

                        if (!result.IsValid || !result.Found)
                        {
                            Log.Logger.Error("{@model}", new
                            {
                                Id = _id,
                                Exception = new Exception(result.DebugInformation),
                                Message = string.Format("can not find join by join._id {0}", _id)
                            });
                            return;
                        }

                        Policy.HandleResult<IUpdateResponse<EventJoin>>(r => !r.IsValid)
                            .WaitAndRetry(4, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                                (a, b, c, d) =>
                                {
                                    if (c > 2)
                                    {
                                        Log.Logger.Error("{@model}", new
                                        {
                                            Id = _id,
                                            Exception = new Exception(a.Result.DebugInformation)
                                        });
                                    }
                                }).Execute(() =>
                                {
                                    var response = _elasticClient.Update<EventJoin, object>(
                                        new DocumentPath<EventJoin>(_id),
                                        t => t
                                            .Index(_settings.ElasticsearchIndex)
                                            .Type(_settings.ElasticsearchType)
                                            .DocAsUpsert()
                                            .Doc(new
                                            {
                                                ExtraFieldsForES = join.ExtraFieldsForES
                                            }));

                                    return response;
                                }
                            );
                    }
                }
                catch (Exception e)
                {
                    Log.Logger.Error("{@model}", new
                    {
                        Id = _id,
                        Exception = e
                    });
                }
            }, true);
        }

        public void SyncByEventId()
        {
            RabbitHub.Subscribe<string>(_settings.SyncJoinByEventIdQueue, (eventId) =>
            {
                try
                {
                    Log.Logger.Information(string.Format("received eventId [{0}]", eventId));

                    int pageIndex = 1;
                    int pageSize = 1000;

                    var filter = Builders<EventJoin>.Filter.Eq("EventId", long.Parse(eventId));
                    var projection =
                        new BsonDocument {{"_id", 1}, {"JoinId", 1}, {"EventId", 1}, {"ExtraFieldsForES", 1}};
                    while (true)
                    {
                        var joins = _mongoCollection.Find(filter).Project<EventJoin>(projection)
                            .Skip((pageIndex - 1) * pageSize).Limit(pageSize).ToList();
                        if (!joins.Any())
                        {
                            RabbitHub.Setup(_settings.WriteSynchronizedTagExchange, ExchangeType.Fanout,
                                _settings.WriteSynchronizedTagQueue);

                            RabbitHub.Broadcast<string>(_settings.WriteSynchronizedTagExchange, eventId);

                            break;
                        }

                        var result = Policy.HandleResult<List<EventJoin>>(r => r.Any()).WaitAndRetry(4,
                                retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (a, b, c, d) =>
                                {
                                    var missingRecords = a.Result;
                                    var _idArray = missingRecords.Select(t => ObjectId.Parse(t._id));
                                    var _fullFilter = Builders<BsonDocument>.Filter.In("_id", _idArray);
                                    var _projection = Builders<BsonDocument>.Projection.Exclude("ExtraFields")
                                        .Exclude("ExtraFieldsForES");
                                    var _missingDocs = _fullMongoCollection.Find(_fullFilter).Project(_projection)
                                        .ToList()
                                        .Select(t => t.ToDictionary()).ToList();

                                    var flushKeys = new List<string>() {"_id", "__v"};
                                    var _response = _elasticClient.Bulk(bulk => bulk.IndexMany(_missingDocs,
                                        (m, e) => m
                                            .Index(_settings.ElasticsearchIndex)
                                            .Type(_settings.ElasticsearchType)
                                            .Id(e["_id"].ToString())
                                            .Document(e.Where(t => !flushKeys.Contains(t.Key))
                                                .ToDictionary(t => t.Key, t => t.Value))));

                                    if (c > 2 && !_response.IsValid)
                                    {
                                        Log.Logger.Error("{@model}", new
                                        {
                                            EventId = eventId,
                                            PageSize = pageSize,
                                            PageIndex = pageIndex,
                                            EventJoins = a.Result,
                                            Exception = new Exception(_response.DebugInformation)
                                        });
                                    }
                                })
                            .Execute(() =>
                            {
                                var missingRecords = new List<EventJoin>()
                                {
                                    new EventJoin()
                                };

//                                var many = _elasticClient.Search<EventJoin>(s =>
//                                    s.Index(_settings.ElasticsearchIndex).Type(_settings.ElasticsearchType).Query(q =>
//                                            q.Terms(t => t.Field(e => e._id).Terms(joins.Select(e => e._id))))
//                                        .Source(t => t.Includes(i => i.Field(e => e._id))).Size(pageSize));

                                var many = _elasticClient.Search<EventJoin>(s =>
                                    s.Index(_settings.ElasticsearchIndex).Type(_settings.ElasticsearchType).Query(q =>
                                            q.Bool(b => b.Must(
                                                Query<EventJoin>.Term(t =>
                                                    t.Field(e => e.EventId).Value(long.Parse(eventId))),
                                                Query<EventJoin>.Terms(t =>
                                                    t.Field(e => e._id).Terms(joins.Select(e => e._id)))
                                            ))
                                        )
                                        .Source(t => t.Includes(i => i.Field(e => e._id))).Size(pageSize));

                                if (many.IsValid)
                                {
                                    var docs = many.Hits.Select(t => t.Id).ToList();
                                    missingRecords = joins.Where(t => !docs.Exists(e => e == t._id)).ToList();
                                }

                                return missingRecords;
                            });

                        if (result.Any())
                        {
                            Log.Logger.Error("{@model}", new
                            {
                                EventId = eventId,
                                PageSize = pageSize,
                                PageIndex = pageIndex,
                                EventJoins = result,
                                Message = "can not find these joins in elasticsearch"
                            });
                            continue;
                        }

                        Policy.Handle<Exception>().OrResult<IBulkResponse>(r => !r.IsValid)
                            .WaitAndRetry(4, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                                (a, b, c, d) =>
                                {
                                    if (c > 2)
                                    {
                                        Log.Logger.Error("{@model}", new
                                        {
                                            EventId = eventId,
                                            Exception = new Exception(a.Result.DebugInformation)
                                        });
                                    }
                                }).Execute(() =>
                                {
                                    var response = _elasticClient.Bulk(b => b.UpdateMany<EventJoin, object>(joins,
                                        (u, e) =>
                                            u.Index(_settings.ElasticsearchIndex)
                                                .Type(_settings.ElasticsearchType)
                                                .Id(e._id)
                                                .DocAsUpsert()
                                                .Doc(new
                                                {
                                                    ExtraFieldsForES = e.ExtraFieldsForES
                                                })));

                                    return response;
                                }
                            );

                        pageIndex++;
                    }
                }
                catch (Exception e)
                {
                    Log.Logger.Error("{@model}", new
                    {
                        EventId = eventId,
                        Exception = e
                    });
                }
            }, true);
        }
    }
}